package demoflink.sink;


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;


public class SinkKeyKafka {



    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //并行度为1
        env.setParallelism(1);
        //如果是精准一次必须要开启
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        DataStreamSource<String> source = env.socketTextStream("192.168.101.121", 7777);
        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.101.121:9092,192.168.101.122:9092,192.168.101.123:9092")
                .setRecordSerializer(
                        new KafkaRecordSerializationSchema<String>() {
                            @Nullable
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(String s, KafkaSinkContext kafkaSinkContext, Long aLong) {
                                String[] split = s.split(",");
                                byte[] key = split[0].getBytes(StandardCharsets.UTF_8);
                                byte[] value = s.getBytes(StandardCharsets.UTF_8);
                                return new ProducerRecord<>("demo-flink",key,value);
                            }
                        }
                )
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                //如果是精准一次必须设置事务前缀
                .setTransactionalIdPrefix("demo-flink-")
                //如果是精准一次必须设置事务超时时间 时间应该大于checkpoint 小于max 15分钟
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10*60*1000 + "")
                .build();
        source.sinkTo(sink);
        env.execute();
    }
}
